package com.xframe.event.bus.core.msg;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;

public class FanoutReceiveMsg implements IReceiveMsg {

    protected Connection msgConn;
    protected Channel curChannel = null;
    protected String queueName;
    protected int qos = 1;
    protected boolean autoAck = true;
    protected IMsgProc msgProc = null;
    protected String exchange;

    @Override
    public Channel getCurChannel() {
        return curChannel;
    }

    @Override
    public IMsgProc getMsgProc() {
        return msgProc;
    }

    @Override
    public Connection getMsgConn() {
        return msgConn;
    }

    @Override
    public int getQos() {
        return qos;
    }

    @Override
    public boolean isAutoAck() {
        return autoAck;
    }

    public FanoutReceiveMsg(Connection msgConn, IMsgProc msgProc, String queueName,String exchange){
        this.msgConn = msgConn;
        this.msgProc = msgProc;
        this.exchange = exchange;
        this.queueName = queueName;
    }


    protected BuiltinExchangeType getBuiltinExchangeType(){
        return BuiltinExchangeType.FANOUT;
    }

    @Override
    public void init() {
        try{
            curChannel = msgConn.createChannel();
            curChannel.exchangeDeclare(this.exchange, getBuiltinExchangeType(), true);
            curChannel.queueDeclare(this.queueName, true, false, false, null);
            curChannel.queueBind(this.queueName, this.exchange, "");
            DefaultConsumer defaultConsumer = new DefaultReceiveConsumer(this);
            curChannel.basicConsume(queueName, this.autoAck, defaultConsumer);
        }catch (Exception exception){
            exception.printStackTrace();
        }
    }
}
